package cn.edu.flink.scala.tutorial.window

import cn.edu.flink.scala.tutorial.source.{ClickSource, Event}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

// 定义统计输出的结果数据结构
case class UrlViewCount(url: String, count: Long, windowStart: Long, windowEnd: Long)

object UrlViewCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val eventDS = env.addSource(new ClickSource)
      .assignAscendingTimestamps(_.timestamp)  // WatermarkStrategy.forMonotonousTimestamps() 的替代

    eventDS
      .keyBy(_.url)
      .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
      .aggregate(new UrlViewCountAgg, new UrlViewCountResult)  // 增量聚合
      .print()

    env.execute("UrlViewCount")
  }

  class UrlViewCountAgg extends AggregateFunction[Event, Long, Long] {
    override def createAccumulator(): Long = 0L

    override def add(value: Event, accumulator: Long): Long = accumulator + 1

    override def getResult(accumulator: Long): Long = accumulator

    override def merge(a: Long, b: Long): Long = ???
  }

  class UrlViewCountResult extends ProcessWindowFunction[Long, UrlViewCount, String, TimeWindow] {
    override def process(url: String, context: Context, elements: Iterable[Long], out: Collector[UrlViewCount]): Unit = {
      // 提取需要的数据
      val count = elements.iterator.next()
      val start = context.window.getStart
      val end = context.window.getEnd

      // 输出数据
      out.collect(UrlViewCount(url, count, start, end))
    }
  }
}
